package com.tc.vms;

import com.tc.vms.consumer.ConsumeFromWhere;
import com.tc.vms.message.MessageConst;
import com.tc.vms.producer.MsgMode;
import com.tc.vms.producer.SendMode;
import com.tc.vms.producer.ZipMode;

/**
 * Service-Level Agreement, set features for service
 */
public class SLA {
    private int msgMaxSize = MessageConst.MESSAGE_MAX_SIZE;
    private int asyncQueueSize = 10000;
    private int batchCount = 200;
    private int retryTimesWhenSend = 3;
    private int confirmTimeoutInMs = 3 * 1000;
    private boolean isFailFast = true;
    private SendMode sendMode = SendMode.SYNC;
    private MsgMode msgMode = MsgMode.NORMAL_MESG;
    private ZipMode zipMode = ZipMode.NONE;
    private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.FROM_LARGEST;

    public static final RuntimeException ILLEGAL_ARGUMENT = new RuntimeException("参数必须大于0");

    public SLA() {
    }

    public SLA(final SLA sla) {
        this.msgMaxSize = sla.msgMaxSize;
        this.asyncQueueSize = sla.asyncQueueSize;
        this.batchCount = sla.batchCount;
        this.retryTimesWhenSend = sla.retryTimesWhenSend;
        this.confirmTimeoutInMs = sla.confirmTimeoutInMs;
        this.sendMode = sla.sendMode;
        this.msgMode = sla.msgMode;
        this.zipMode = sla.zipMode;
        this.consumeFromWhere = sla.consumeFromWhere;
        this.isFailFast = sla.isFailFast;
    }

    public SLA setSendMode(SendMode sm) {
        this.sendMode = sm;
        return this;
    }

    public ConsumeFromWhere consumeFromWhere() {
        return consumeFromWhere;
    }

    public SLA setConsumeFromWhere(ConsumeFromWhere consumeFromWhere) {
        this.consumeFromWhere = consumeFromWhere;
        return this;
    }

    /**
     * The maximum number of unsent messages that can be queued up the producer
     * when using async mode before either the producer must be blocked or data
     * must be dropped,which works on kafka so far.Default value is 10000
     *
     * @param size
     * @return
     */
    public SLA setAsyncQueueSize(int size) {
        if (size > 0) {
            this.asyncQueueSize = size;
        }
        return this;
    }

    /**
     * The number of messages to send in one batch when using async mode. The
     * producer will wait until either this number of messages are ready to send
     * or queue.buffer.max.ms is reached. which works in kafka
     *
     * @param count
     * @return
     */
    public SLA setBatchCount(int count) {
        if (count > 0) {
            this.batchCount = count;
        }
        return this;
    }


    /**
     * used in sending message by kafka, select different compressed mode to compress message
     * to improve transferring performance on network
     *
     * @return
     */
    public ZipMode zipMode() {
        return zipMode;
    }

    public SLA setZipMode(ZipMode zipMode) {
        this.zipMode = zipMode;
        return this;
    }

    /**
     * The number of messages to send in one batch when using async mode. The
     * producer will wait until either this number of messages are ready to send
     * or queue.buffer.max.ms is reached.
     *
     * @return
     */
    public long batchCount() {
        return this.batchCount;
    }

    /**
     * Sets the VMS's message size supported size.which throws exception once
     * out of the limition. Default value is 900k.
     *
     * @param msgMaxSize
     * @return
     */
    public SLA setMessageMaxSize(int msgMaxSize) {
        if (msgMaxSize > 0) {
            this.msgMaxSize = msgMaxSize;
        }
        return this;
    }

    public int confirmTimeoutInMs() {
        return confirmTimeoutInMs;
    }

    public SLA setConfirmTimeoutInMs(int confirmTimeoutInMs) {
        if (confirmTimeoutInMs > 0) {
            this.confirmTimeoutInMs = confirmTimeoutInMs;
        }
        return this;
    }

    public long messageMaxSize() {
        return msgMaxSize;
    }

    public long asyncQueueSize() {
        return this.asyncQueueSize;
    }

    public SendMode sendMode() {
        return this.sendMode;
    }

    public int retryTimesWhenSend() {
        return retryTimesWhenSend;
    }

    public SLA setRetryTimesWhenSend(int retryTimesWhenSend) {
        if (retryTimesWhenSend < 0) {
            throw ILLEGAL_ARGUMENT;
        }

        if (retryTimesWhenSend >= 0) {
            this.retryTimesWhenSend = retryTimesWhenSend;
        }
        return this;
    }

    public boolean isFailFast() {
        return isFailFast;
    }

    public SLA setIsFailFast(boolean isFailFast) {
        this.isFailFast = isFailFast;
        return this;
    }

    public MsgMode msgMode() {
        return msgMode;
    }

    public SLA setMsgMode(MsgMode msgMode) {
        this.msgMode = msgMode;
        return this;
    }

    public SLA cloneSLA() {
        return new SLA(this);
    }
}
